package com.djk.daily.kafka.admin;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class AdminSimple {

    public final static String TOPIC_NAME = "djk_topic";

    /**
     * 设置AdminClient
     * @return
     */
    public static AdminClient adminClient(){
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"101.35.134.133:9092");

        AdminClient adminClient = AdminClient.create(properties);
        return adminClient;
    }

    /**
     * 创建topic实例
     */
    public static void createTopic ()  {
        AdminClient adminClient = adminClient();
        //副本实例
        Short rs = 1;
        NewTopic newTopic = new NewTopic(TOPIC_NAME, 1, rs);
        CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic));
        try {
            TimeUnit.SECONDS.sleep(1L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("CreateTopicsResult = " + topics);
    }

    /**
     * 获取topic列表
     */
    public static void topicList() throws Exception {
        AdminClient adminClient = adminClient();
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        //是否查看Internal
        listTopicsOptions.listInternal(true);
        ListTopicsResult listTopicsResult = adminClient.listTopics(listTopicsOptions);
//        ListTopicsResult listTopicsResult = adminClient.listTopics();
        Set<String> names = listTopicsResult.names().get();

        //打印names
        names.stream().forEach(System.out::println);
    }

    /**
     * 删除Topic
     * @throws Exception
     */
    public static void deleteTopic() throws Exception {
        AdminClient adminClient = adminClient();

        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(TOPIC_NAME));
        deleteTopicsResult.all().get();
    }

    /**
     * 描述
     * @throws Exception
     */
    public static void describeTopics() throws Exception{
        AdminClient adminClient = adminClient();
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME));
        Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
        Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
        entries.stream().forEach(en->{
            System.out.println("name : " + en.getKey() + ",desc : " + en.getValue());
        });
    }


    /**
     * 查看Config
     * @throws Exception
     */
    public static void describeConfig() throws Exception{
        AdminClient adminClient = adminClient();
        //TODO 集群时会讲到
//        ConfigResource configResource = new ConfigResource(ConfigResource.Type.BROKER, TOPIC_NAME);
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource));
        describeConfigsResult.all().get().entrySet().stream().forEach(en->{
            System.out.println("configResource : " + en.getKey() + ",Config : " + en.getValue());
        });
    }

    /**
     * Topic配置信息修改
     * @throws Exception
     */
    public static void alterConfig() throws Exception{
        AdminClient adminClient = adminClient();
        Map<ConfigResource, Config> configs = new HashMap<>();

        //组织两个参数
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        Config config = new Config(Arrays.asList(new ConfigEntry("preallocate","true")));
        configs.put(configResource,config);
        AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(configs);
        alterConfigsResult.all().get();
    }

    /**
     * 添加Partitions
     *
     * @throws Exception
     */
    public static void incrPartitions(int partitions) throws Exception{
        AdminClient adminClient = adminClient();
        Map<String,NewPartitions> partitionsMap = new HashMap<>();
        NewPartitions newPartitions = NewPartitions.increaseTo(partitions);
        partitionsMap.put(TOPIC_NAME,newPartitions);

        CreatePartitionsResult createPartitionsResult = adminClient.createPartitions(partitionsMap);

        createPartitionsResult.all().get();
    }

    public static void main(String[] args) throws Exception {
//        AdminClient adminClient = AdminSimple.adminClient();
//        System.out.println("adminClient = " + adminClient);
//        createTopic();
//        deleteTopic();
//        topicList();
        describeTopics();
//        alterConfig();
//        describeConfig();
//        incrPartitions(2);
    }
}
